package com.distributeLock;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.curator.framework.recipes.locks.Lease;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class ZKSemaphone {

    public static void main(String[] args) {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient("127.0.0.1:2181", new ExponentialBackoffRetry(1000, 3));

        curatorFramework.start();

        //指定锁路径
        String lockPath = "/zkLockRoot/lock_1";

        //创建信号量，指定同时最大访问数为3个
        InterProcessSemaphoreV2 semaphoreMutex = new InterProcessSemaphoreV2(curatorFramework, lockPath, 3);


        // 生成线程池
        ExecutorService executorService = Executors.newCachedThreadPool();

        Consumer<InterProcessSemaphoreV2> consumer = (InterProcessSemaphoreV2 semaphore) -> {
            try {
                List<Callable<String>> callableList = new ArrayList<>();

                Callable<String> callable = () -> {
                    Lease lease = semaphore.acquire();

                    try {
                        System.out.println(Thread.currentThread() + " acquire semaphore");
                        TimeUnit.MILLISECONDS.sleep(200);
                    } catch (Exception e) {

                    } finally {
                        semaphore.returnLease(lease);

                        System.out.println(Thread.currentThread() + " release semaphore");
                    }

                    return "true";
                };

                for (int i = 0; i < 5; i ++) {
                    callableList.add(callable);
                }

                List<Future<String>> futures = executorService.invokeAll(callableList);
            } catch (Exception e) {

            }
        };

        System.out.println("5个并发线程，信号量测试");

        consumer.accept(semaphoreMutex);

        executorService.shutdown();

        curatorFramework.close();
    }

}
